I've probably missed some places to change, and this may need more testing.
Also, I need guidance on how to fix the ordering issue: the paired-joins require the first join to preserve the input ordering. Currently I've put a hack in distsql_physical_planner.go to set MaintainOrdering to true, but this should really happen earlier in the optimizer when doing transformations.

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @RaduBerinde and @rytaft)

Nice! Can we remove the ConvertSemiToInnerJoin rule now?

The ordering issue is a bit tricky, since I don't think you want to force a sort on the input -- you just want to make sure that rows originating from the same input row don't get separated.

Also, at the moment, we don't even take into account the need to maintain ordering in our cost model for lookup joins. So I'm not sure you can do much better than the hack you've added (although we could potentially move the hack into the execbuilder). Anything we add in the optimizer to enforce this ordering will probably add some unintended overhead during execution.

@RaduBerinde do you think there is a way to do this cleanly in the optimizer?

Reviewed 15 of 15 files at r1.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @RaduBerinde and @sumeerbhola)

pkg/sql/opt/xform/join_funcs.go, line 361 at r1 (raw file):

In some cases

I'd make clear that this only applies for semi, anti, and some left joins

pkg/sql/opt/xform/join_funcs.go, line 363 at r1 (raw file):

		// In some cases, this lower-upper pair of joins is further specialized
		// into paired-joins where we refer to the lower as first and upper as
		// second. The paired-joins use a continuation column in the first join.

I think this could use more detail. It would be great if you can update the function comment to describe this case. Might want to separate it into "case 3".

pkg/sql/opt/xform/join_funcs.go, line 455 at r1 (raw file):

			indexJoin.IsSecondJoinInPairedJoiner = true
		}

Is it worth putting the rest of this into a helper function? Seems like this is basically identical to the code in GenerateInvertedJoins.

pkg/sql/opt/xform/join_funcs.go, line 468 at r1 (raw file):

		// lookup join).
		outputCols := indexJoin.Cols.Intersection(indexJoin.Input.Relational().OutputCols)
		if outputCols.SubsetOf(input.Relational().OutputCols) {

[nit] to be consistent with the rest of this function, use inputProps.OutputCols instead of input.Relational().OutputCols.

TFTR!
Do we have any existing benchmarks with non-covering indices that I can try this on, or some execution benchmarks that I could extend?

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @RaduBerinde and @sumeerbhola)

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @RaduBerinde and @rytaft)

pkg/sql/opt/xform/join_funcs.go, line 361 at r1 (raw file):

Done

pkg/sql/opt/xform/join_funcs.go, line 363 at r1 (raw file):

Done.

pkg/sql/opt/xform/join_funcs.go, line 455 at r1 (raw file):

Good point. Done

pkg/sql/opt/xform/join_funcs.go, line 468 at r1 (raw file):

Done.

pkg/sql/opt/xform/join_funcs.go, line 161 at r2 (raw file):

//     and then use a lookup join to retrieve column z. The "index join" (top
//     LookupJoin) will produce columns a,b,c,x,y,z; the lookup columns are just x
//     (the original index join produced a,b,c,x,y).

I changed this comment since it seemed incorrect, though maybe I misunderstood it. PTAL.

Reviewed 2 of 3 files at r2, 4 of 4 files at r3.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @RaduBerinde and @sumeerbhola)

pkg/sql/opt/xform/join_funcs.go, line 455 at r1 (raw file):

Can you call the function from GenerateInvertedJoins too?

pkg/sql/opt/xform/join_funcs.go, line 161 at r2 (raw file):

I think this should still say "the lookup columns are just z", but the other changes you made look right to me.

Also might want to change "original index join" to say "original lookup join".

pkg/sql/opt/xform/join_funcs.go, line 174 at r3 (raw file):

//     joins, but paired-joins explicitly know their role in the pair and
//     behave accordingly.
//     For example, using the same tables in the example for case 2:

[nit] add an empty line above this one

pkg/sql/opt/xform/join_funcs.go, line 175 at r3 (raw file):

//     behave accordingly.
//     For example, using the same tables in the example for case 2:
//      SELECT * FROM abc JOIN xyz ON a=y AND b=z

maybe change this to LEFT JOIN so it would actually use paired joins

pkg/sql/opt/xform/join_funcs.go, line 484 at r3 (raw file):

}

// addIndexJoinAsSecondJoin is a helper function used when constructing two

[nit] this should match the name of the function. But I kind of like this name better than the function name... I'd remove the "ToMemo" from the function name, since I think that's implied.

Also, I need guidance on how to fix the ordering issue: the paired-joins require the first join to preserve the input ordering. Currently I've put a hack in distsql_physical_planner.go to set MaintainOrdering to true, but this should really happen earlier in the optimizer when doing transformations.

The ordering issue is a bit tricky, since I don't think you want to force a sort on the input -- you just want to make sure that rows originating from the same input row don't get separated.

I don't think MaintainOrdering=true will force a sort on the input -- it switches the joinReader to use joinReaderOrderingStrategy that needs to buffer the looked up rows (unlike joinReaderNoOrderingStrategy) so it can produce all the output rows for an input row in-order, which may result in spilling to disk, so can be more expensive.

Also, at the moment, we don't even take into account the need to maintain ordering in our cost model for lookup joins. So I'm not sure you can do much better than the hack you've added (although we could potentially move the hack into the execbuilder). Anything we add in the optimizer to enforce this ordering will probably add some unintended overhead during execution.

@RaduBerinde do you think there is a way to do this cleanly in the optimizer?

This ordering issue is the main open issue. I'd like a cleaner solution.
@RaduBerinde do you have any thoughts on this?

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @RaduBerinde and @rytaft)

pkg/sql/opt/xform/join_funcs.go, line 455 at r1 (raw file):

It does below. Maybe I misunderstood your comment.

pkg/sql/opt/xform/join_funcs.go, line 161 at r2 (raw file):

The term "lookup columns" seems poorly defined. In LookupJoinPrivate there are comments like:
"set of lookup columns retrieved through lookup" (which suggests z), and "lookup columns form a key in the table" (which suggests x), and the JoinReaderSpec comment "(lookup columns is [0])" also suggest x.
I am changing it back to z for now.

And changed to "original lookup join".

pkg/sql/opt/xform/join_funcs.go, line 174 at r3 (raw file):

Done.

pkg/sql/opt/xform/join_funcs.go, line 175 at r3 (raw file):

Done.

pkg/sql/opt/xform/join_funcs.go, line 484 at r3 (raw file):

Done. I changed one and forgot to change the other.

I don't think MaintainOrdering=true will force a sort on the input

Right, I agree. What I meant was that I think if you try to maintain ordering from the perspective of the optimizer, it will force a sort on the input. I don't think we have a way for the optimizer to set MaintainOrdering=true unless there is an explicit ordering.

@RaduBerinde might have some insight, but he's out of office until next week.

Reviewed 1 of 1 files at r4.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @RaduBerinde and @sumeerbhola)

pkg/sql/opt/xform/join_funcs.go, line 455 at r1 (raw file):

Sorry! Not sure how I missed that.

pkg/sql/opt/xform/join_funcs.go, line 161 at r2 (raw file):

Yea, I agree it's confusing. You could also change "lookup columns" to say "the columns retrieved from the primary index". But up to you -- LGTM otherwise.

If I understand correctly, the paired joins are guaranteed to be paired 1-to-1 with no data shuffling between the two stages, correct? If that's the case, we can store the continuation column in both the upper and the lower lookup join and have the upper join always require an ordering on the continuation column. The code in lookupOrIndexJoinCanProvideOrdering would be changed to reflect that a lookup join can always provide an ordering on the continuation column. We're lying to distsql that we have an ordering on that column, but that's ok since we don't use ordered synchronizers if there is a single input stream.

I don't think this would be better than what you have here though (seems to introduce more confusion). If we care about any costing differences, we can directly take into account flags like IsFirstJoinInPairedJoiner in the coster.

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @sumeerbhola)

pkg/sql/opt/ops/relational.opt, line 343 at r4 (raw file):

    # IsFirstJoinInPairedJoiner is true if this is the first join of a
    # paired-joiner used for left joins.
    IsFirstJoinInPairedJoiner bool

[nit] I would use "lower" and "upper" everywhere, since it's not obvious which is first and which is second (depends if you think about the plan tree or the flow of data).

(trying to translate this into something I understand, given my limited understanding of the optimizer code)

If I understand correctly, the paired joins are guaranteed to be paired 1-to-1 with no data shuffling between the two stages, correct?

Yes, we want them to stay paired with no data shuffling. If data is shuffled between the two, the result will be incorrect. The existing if-block in lookupOrIndexJoinCanProvideOrdering

If that's the case, we can store the continuation column in both the upper and the lower lookup join and have the upper join always require an ordering on the continuation column. The code in lookupOrIndexJoinCanProvideOrdering would be changed to reflect that a lookup join can always provide an ordering on the continuation column.

The continuation column is a bool, and we expect each input to produce a sequence of (false, true*). Actually sorting by it would be incorrect.
If the upper join also provided a continuation column, I am assuming we would then need to add a project to remove it, yes?
Also, I'm unclear on how lookupOrIndexJoinCanProvideOrdering can claim to provide ordering on the continuation column, given that it is just answering true/false to a question it is being asked regarding the OrderingChoice parameter. Who would adjust the OrderingChoice parameter to include the continuation col? And there could be other expressions in the same memo group as the paired-joins that don't provide the continuation column -- I'm not clear how that would work.

We're lying to distsql that we have an ordering on that column, but that's ok since we don't use ordered synchronizers if there is a single input stream.

I am assuming you mean that each stream from a lower joinReader will have a corresponding upper joinReader, so there won't be an OrderedSynchronizer between the two. Correct?
But what about above the upper joinReader, since it is also claiming to provide an ordered continuation column -- couldn't there be an OrderedSynchronizer there to merge streams from multiple upper joinReaders?

I don't think this would be better than what you have here though (seems to introduce more confusion). If we care about any costing differences, we can directly take into account flags like IsFirstJoinInPairedJoiner in the coster.

Are you suggesting that we would change computeIndexLookupJoinCost in coster.go to look at whether memo.LookupJoinExpr has IsFirstJoinInPairedJoiner set, and if so change its costs knowing that it will respect input ordering? I can add a TODO there, for when the optimizer starts taking ordering cost into account for lookup joins. Or if you all already have a cost model in mind, I can add it now.

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @sumeerbhola)

I am not suggesting that the upper lookup join produces a continuation column. I am suggesting that it knows the ID of the continuation column from the lower join, so that it can always require an ordering on that column from its child (that would happen in lookupOrIndexJoinBuildChildReqOrdering); then the canProvide in the child would know that it can always provide that ordering. I understand that the column isn't sorted in reality, we would just be pretending from the optimizer side. It definitely doesn't fix any core issue (and in fact, we will probably need special code to know that this special ordering can't be enforced with a sort).

In any case, we would still be relying on two things:

• that the optimizer doesn't further transform the lower lookup join; this is the case because we don't have any rules that start from a lookup join, and if we did they would need to know about a possible continuation column anyway; in addition, the hack you added to never allow a sort in-between).
• that distsql doesn't shuffle data between the two (this code https://github.com/cockroachdb/cockroach/blob/master/pkg/sql/distsql_physical_planner.go#L2103)

We could upgrade the continuation column to an int64 with an increasing "group" index. Then we would actually have a true ordering on this column (and it wouldn't be incorrect to enforce it by sorting). If we mix in a host ID or random value, we could in principle allow for multiple lower joiners to feed in a single upper joiner with an ordered synchronizer. But this feels unnecessary in the current implementation.

I would just go with your current solution in the PR for now. Longer term, we should brainstorm on what the missing concept is. The point of the continuation column is to be able to group the results for a single input row, is this correct? I think we could achieve that in a much more elegant fashion by wrapping the input in an Ordinality and then plumbing through that ordinality column. We would have a true ordering on it and we wouldn't need to create any special columns. Ordinality doesn't get distributed, but if we care about that, we could add a DistributedOrdinality variant that adds a node id or similar to the output value to prevent collisions (we could use that in some other cases as well).

And there could be other expressions in the same memo group as the paired-joins that don't provide the continuation column

Oh.. I assumed that these lookup joins produce the continuation column as an output column (which would mean they can't be in the same group with expressions that don't produce that column). It looks like they don't, which is pretty sketchy. Until they do, we can't do anything like requiring an ordering on that column.

Are you suggesting that we would change computeIndexLookupJoinCost in coster.go to look at whether memo.LookupJoinExpr has IsFirstJoinInPairedJoiner set, and if so change its costs knowing that it will respect input ordering?

Right, you can just add a TODO.

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @sumeerbhola)

pkg/sql/opt/exec/execbuilder/relational.go, line 1477 at r4 (raw file):

		}
		// Assign the continuation column the next unused value in the map.
		allCols.Set(int(join.ContinuationCol), maxValue+1)

Is it possible for a first paired join to be an anti or semi? Because the column wouldn't be in the map in that case (see the following few lines)

TFTRs!

I'll hold off on merging until we have a solution for #58892

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @sumeerbhola)

I did the rebase, and all the opt tests are passing now. Let's see if any other tests fail. Otherwise, I think this is ready to go. Does anyone from SQL Queries want to take a quick glance at this to make sure I didn't miss something?

Reviewed 16 of 16 files at r5.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @RaduBerinde and @sumeerbhola)

pkg/sql/opt/exec/execbuilder/relational.go, line 1477 at r4 (raw file):

No I think the first join is always either a left or an inner join.

pkg/sql/opt/ops/relational.opt, line 343 at r4 (raw file):

I'm going to leave this as-is for now since first / second is already used throughout the codebase and this would be a big change.

Reviewed 1 of 1 files at r6.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @RaduBerinde and @sumeerbhola)

Did a quick pass. Probably it'd be good to get an approval from @RaduBerinde.

Reviewed 15 of 16 files at r5, 1 of 1 files at r6.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @RaduBerinde, @rytaft, and @sumeerbhola)

pkg/sql/lookup_join.go, line 69 at r6 (raw file):

	// At most one of is{First,Second}JoinInPairedJoiner can be true.
	// IsFirstJoinInPairedJoiner can be true only if reqOrdering asks the join

Doesn't this comment contradict the TODO added in the distsql_physical_planner?

pkg/sql/opt/exec/execbuilder/relational.go, line 1477 at r4 (raw file):

nit: maybe add this observation as a comment.

pkg/sql/opt/xform/join_funcs.go, line 674 at r6 (raw file):

				// The lowerJoinType continues to be LeftJoinOp.
			}
			// We have already set pairedJoins=true for SemiJoin,AntiJoin earlier,

nit: add a space before AntiJoin.

pkg/sql/opt/exec/execbuilder/testdata/lookup_join, line 774 at r6 (raw file):

# Left anti-join with ON filter on non-covering index
query T
EXPLAIN (VERBOSE) SELECT small.c FROM small WHERE NOT EXISTS(SELECT 1 FROM large WHERE small.c = large.b AND large.d < small.a)

nit: this query doesn't seem to match to what is added in the logic tests above.

TFTR! Updated.

Reviewed 3 of 3 files at r7, 1 of 1 files at r8.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @RaduBerinde, @sumeerbhola, and @yuzefovich)

pkg/sql/lookup_join.go, line 69 at r6 (raw file):

Good point. Fixed.

pkg/sql/opt/exec/execbuilder/relational.go, line 1477 at r4 (raw file):

I added an assertion

pkg/sql/opt/xform/join_funcs.go, line 674 at r6 (raw file):

Done.

pkg/sql/opt/exec/execbuilder/testdata/lookup_join, line 774 at r6 (raw file):

Done.

Cool!

Reviewed 11 of 16 files at r5, 1 of 1 files at r6, 3 of 3 files at r7, 1 of 1 files at r8.
Reviewable status: complete! 1 of 0 LGTMs obtained (waiting on @RaduBerinde, @rytaft, @sumeerbhola, and @yuzefovich)

pkg/sql/opt_exec_factory.go, line 698 at r8 (raw file):

	n.columns = pred.cols
	if isFirstJoinInPairedJoiner {
		n.columns = append(n.columns, colinfo.ResultColumn{Name: "cont", Typ: types.Bool})

It seems dangerous to assume the continuation column is always the last column. Is there no way to use the continuation column in the LookupJoinPrivate to make this more robust?

pkg/sql/opt/memo/check_expr.go, line 250 at r8 (raw file):

				}
			default:
				panic(errors.AssertionFailedf("lookup paired-join is paired with %T", t.Input))

nit: you could also add validation for the continuation columm

pkg/sql/opt/ops/relational.opt, line 343 at r4 (raw file):

Can you expand these comments to clarify which join is first here which is second?

pkg/sql/opt/ops/relational.opt, line 416 at r8 (raw file):

    # ContinuationCol is the column ID of the continuation column when
    # IsFirstJoinInPairedJoiner is true.

nit: an explanation of this continuation column might be nice here

pkg/sql/opt/xform/join_funcs.go, line 204 at r8 (raw file):

//     where cont is the continuation column used to group together rows that
//     correspond to the same original a,b,c. The second join will evaluate
//     b=z and produce columns a,b,c,x,y,z. A similar approach works for

Might be more clear to say: "The second join will fetch z from the primary index, evaluate b=z, and produce ...". I was confused because I originally thought it meant the second lookup join will have key columns b and z, which wouldn't be possible without an index on z.

pkg/sql/opt/xform/testdata/rules/join, line 2339 at r8 (raw file):

 │    ├── columns: m:1 n:2 a:6 b:7 abcd.rowid:9 continuation:12
 │    ├── key columns: [1] = [6]
 │    ├── fd: (9)-->(6,7,12)

nit: it might be helpful to print first join in paired joiner and the continuation column in the expression tree

TFTR!

Reviewed 8 of 8 files at r9.
Reviewable status: complete! 0 of 0 LGTMs obtained (and 1 stale) (waiting on @mgartner, @sumeerbhola, and @yuzefovich)

pkg/sql/opt_exec_factory.go, line 698 at r8 (raw file):

I don't really see a way to do that here. We already have that logic inside the execbuilder.

pkg/sql/opt/exec/execbuilder/relational.go, line 1477 at r4 (raw file):

Moved the assertion to check_expr

pkg/sql/opt/exec/execbuilder/relational.go, line 1769 at r8 (raw file):

				join.JoinType.String(),
			)
		}

Note that I moved this assertion to check_expr

pkg/sql/opt/memo/check_expr.go, line 250 at r8 (raw file):

Done.

pkg/sql/opt/ops/relational.opt, line 343 at r4 (raw file):

Done.

pkg/sql/opt/ops/relational.opt, line 416 at r8 (raw file):

Done.

pkg/sql/opt/xform/join_funcs.go, line 204 at r8 (raw file):

Done.

pkg/sql/opt/xform/testdata/rules/join, line 2339 at r8 (raw file):

Done.

bors r+

Reviewable status: complete! 0 of 0 LGTMs obtained (and 1 stale) (waiting on @mgartner, @sumeerbhola, and @yuzefovich)

Reviewed 8 of 8 files at r9.
Reviewable status: complete! 0 of 0 LGTMs obtained (and 1 stale)

pkg/sql/opt/exec/execbuilder/relational.go, line 1769 at r8 (raw file):

I'm not sure it matters, but perviously this assertion would run for release builds, but in check_expr it will only run for test builds.